package com.mjf.app.function;

import com.alibaba.fastjson.JSONObject;
import com.mjf.common.GmallConfig;
import com.mjf.utils.DimUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

/**
 * 自定义实现通过 phoenix 写入数据到 HBase
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    private Connection conn;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
        conn.setAutoCommit(true);   // 设置自动提交
    }

    /**
     * 将数据插入 phoenix(HBase)
     * @param value
     * @param context
     * @throws Exception
     */
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        // 获取表名
        String tableName = value.getString("sinkTable");
        // 获取数据
        JSONObject data = value.getJSONObject("after");

        // 获取 SQL 语句
        String upsertSql = getUpsertSql(tableName, data);
        System.out.println(upsertSql);

        PreparedStatement preparedStatement = null;

        try {
            // 预编译 SQL
            preparedStatement = conn.prepareStatement(upsertSql);

            if ("update".equals(value.getString("type"))) {
                // 如果当前为数据更新操作，则先删除Redis中缓存的数据
                // 如果当心在redis删除旧数据之后，HBase插入新数据之前这段时间会有旧数据重新写入到redis。可以将下面的删除换成修改
                DimUtil.delRedisDimInfo(tableName.toUpperCase(), data.getString("id").toUpperCase());
            }

            // 执行插入操作
            preparedStatement.executeUpdate();
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            if (preparedStatement != null) {
                preparedStatement.close();
            }
        }
    }

    /**
     * 获取 upsert 语句
     * @param sinkTable 表名
     * @param data {"id":"1", "tm_name":"华为"}
     * @return upsert into db.tn (id, tm_name) values('xxx', 'xxx')
     */
    private String getUpsertSql(String sinkTable, JSONObject data) {
        Set<String> keys = data.keySet();
        Collection<Object> values = data.values();
        return "upsert into "
                + GmallConfig.HBASE_SCHEMA
                + "."
                + sinkTable
                + " ("
                + StringUtils.join(keys, ",")
                + ") "
                + "values('"
                + StringUtils.join(values, "','")
                + "')"
                ;
    }
}
